package com.hk.urils;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;

import java.util.Properties;

public class KafkaSource {

    public static void configureSasl(Properties prop) {
        //如果用-D或者其它方式设置过，这里不再设置
        if (null == System.getProperty("java.security.auth.login.config")) {
            //请注意将XXX修改为自己的路径
            //这个路径必须是一个文件系统可读的路径，不能被打包到jar中
            System.setProperty("java.security.auth.login.config", prop.getProperty("java.security.auth.login.config"));
        }
    }

    public static Properties getKafkaParam(Properties prop){

        Properties consumerProp = new Properties();
//        consumerProp.setProperty("bootstrap.servers", prop.getProperty("bootstrap.servers"));
//        consumerProp.setProperty("zookeeper.connect", prop.getProperty("zookeeper.connect"));
//        consumerProp.put("group.id", prop.getProperty("group.id"));
        // 设置非自动提交offset
        consumerProp.put("enable.auto.commit",prop.getProperty("enable.auto.commit"));
        consumerProp.put("auto.commit.interval.ms", prop.getProperty("auto.commit.interval.ms"));
        // 当有提交的offset时，从提交的位置开始继续消费，否则从头开始
        consumerProp.put("auto.offset.reset", "earliest");
//        consumerProp.put("auto.offset.reset",prop.getProperty("auto.offset.reset"));
//        consumerProp.put("key.deserizlizer","org.apache.kafka.common.serialization.StringDeserializer");
//        consumerProp.put("value.deserizlizer","org.apache.kafka.common.serialization.StringDeserializer");

        //设置接入点，即控制台的实例详情页显示的“SSL接入点”
        consumerProp.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getProperty("bootstrap.servers"));
        //设置SSL根证书的路径，请记得将XXX修改为自己的路径
        //与SASL路径类似，该文件也不能被打包到jar中
        consumerProp.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, prop.getProperty("ssl.truststore.location"));
        //根证书store的密码，保持不变
        consumerProp.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
        //接入协议，目前支持使用SASL_SSL协议接入
        consumerProp.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        //SASL鉴权方式，保持不变
        consumerProp.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        //两次poll之间的最大允许间隔
        //请不要改得太大，服务器会掐掉空闲连接，不要超过30000
        consumerProp.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 25000);
        //每次poll的最大数量
        //注意该值不要改得太大，如果poll太多数据，而不能在下次poll之前消费完，则会触发一次负载均衡，产生卡顿
        consumerProp.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        //消息的反序列化方式
        consumerProp.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProp.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // key序列化
        consumerProp.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // value序列化
        consumerProp.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //当前消费实例所属的消费组，请在控制台申请之后填写
        //属于同一个组的消费实例，会负载消费消息
        consumerProp.put(ConsumerConfig.GROUP_ID_CONFIG, "l-gid2");

        return consumerProp;
    }
}
